package com.gugee.ins.data.collector.blogger.service.impl;

import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.ins.common.beanutil.InsBloggerConversion;
import com.gugee.ins.common.constant.InsBloggerMsgTypeCodeEnum;
import com.gugee.ins.common.dal.service.blogger.InsBloggerDao;
import com.gugee.ins.common.dal.service.dic.InsDicCategoryDao;
import com.gugee.ins.common.dal.service.dic.InsDicCategoryEnumDao;
import com.gugee.ins.common.dal.service.spider.SpiderSvc;
import com.gugee.ins.common.dal.spider.InsBloggerSpiderClient;
import com.gugee.ins.common.model.mysql.blogger.InsBlogger;
import com.gugee.ins.common.model.spider.blogger.InsBloggerByIdResponse;
import com.gugee.ins.common.model.spider.blogger.InsBloggerByUsernameResponse;
import com.gugee.ins.data.collector.blogger.config.CollectorBloggerConfig;
import com.gugee.ins.data.collector.blogger.service.InsBloggerService;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ForkJoinPool;

/**
 * insBlogger服务
 * @author Huangxin
 */
@Slf4j
@Service
public class InsBloggerServiceImpl implements InsBloggerService {

    /**
     * 配置
     */
    @Autowired
    CollectorBloggerConfig collectorBloggerConfig;

    /**
     * 线程调度
     */
    @Autowired
    ScheduleTemplate scheduleTemplate;

    /**
     * 爬虫接口
     */
    @Autowired
    InsBloggerSpiderClient insBloggerSpiderClient;

    /**
     * 爬虫接口
     */
    @Autowired
    SpiderSvc spiderSvc;

    /**
     * kafkaTemplate
     */
    @Autowired
    KafkaTemplate<Long,InsBloggerMessage> kafkaTemplateInsBlogger;


    /**
     *
     */
    @Autowired
    InsBloggerDao insBloggerDao;

    /**
     *
     */
    @Autowired
    InsDicCategoryEnumDao insDicCategoryEnumDao;

    /**
     *
     */
    @Autowired
    InsDicCategoryDao insDicCategoryDao;

    /**
     * 线程池
     */
    ForkJoinPool forkJoinPool;

    /**
     * 线程名
     */
    private String workName = "insBg-upd";

    /**
     * 初始化线程池
     */
    @PostConstruct
    public void init() {
        forkJoinPool = ThreadPoolUtils.forkJoinPool(collectorBloggerConfig.getInsBloggerUpdateCorePollSize(), workName);
        scheduleTemplate.attach(workName, forkJoinPool);
    }

    /**
     * 处理insBlogger收入任务
     * @param insBloggerMessages
     */
    @Override
    public void handleInsBloggerUpdate(List<InsBloggerMessage> insBloggerMessages) {
        try {

            /**
             * 本着节约qps宗旨 需对不断更新uid和只需一次uid的消息消费类型做区别对待，
             * 来自推荐流收录逻辑、大播主关联到的播主收录等收录都称为一次uid消息消费类型
             * 非一次uid消息消费类型设置24小时内更新过则跳过,
             */
            List<Long> uids=new ArrayList<>(insBloggerMessages.size());
            List<Long> uidsCollect=new ArrayList<>(insBloggerMessages.size());

            for (InsBloggerMessage msg:insBloggerMessages) {
                if(msg.getTypeCode() ==0 || msg.getTypeCode() == InsBloggerMsgTypeCodeEnum.UID_UPDATE_F_COLLECT.getCode()
                || msg.getTypeCode() == InsBloggerMsgTypeCodeEnum.UID_UPDATE_F_MBLOGGER_RELATION_TIMING.getCode()

                ){
                    uidsCollect.add(msg.getUid());
                }else {
                    uids.add(msg.getUid());
                }
            }
            Set<Long> uidsSet = new HashSet<>(insBloggerMessages.size());
            if(!uids.isEmpty()){
                List<Long> uidsExists = insBloggerDao.findExistUidByUidsWithCollectTime(uids,System.currentTimeMillis()/1000-86400);
                uidsSet.addAll(uidsExists);
            }
            if(!uidsCollect.isEmpty()){
                List<Long> uidsCollectExists = insBloggerDao.findExistUidByUidsWithCollectTime(uidsCollect,1);
                uidsSet.addAll(uidsCollectExists);
            }

            List<InsBloggerMessage> rstMessages=new ArrayList<>(insBloggerMessages.size()-uidsSet.size());
            for (InsBloggerMessage msg:insBloggerMessages) {
                if(!uidsSet.contains(msg.getUid())){
                    rstMessages.add(msg);
                }
            }
            if(rstMessages.isEmpty()){
                return;
            }

            forkJoinPool.submit(() -> rstMessages.parallelStream().forEach(msg -> {
                try {
                    //采集播主关注人数据
                    runUpdateInsBlogger(msg);
                } catch (Exception ex) {
                    ex.printStackTrace();
                    log.error("InsBlogger update error. uid:{}, error :{}" ,
                            msg.getUid(),ex.getMessage());
                }
            })).get();
        } catch (Exception ex) {
            log.error("InsBlogger update  thread error.  {}" ,  ex.getMessage());
        }
    }

    /**
     * 采集更新播主数据
     * @param insBloggerMessages
     */
    public void runUpdateInsBlogger(InsBloggerMessage insBloggerMessages){
        long start = System.currentTimeMillis();
        InsBlogger insBlogger;
        //根据uid爬取信息
        try {
            InsBloggerByIdResponse insBloggerByIdResponse = spiderSvc.getInsBloggerByUid(insBloggerMessages.getUid());
            insBlogger = InsBloggerConversion.mapper(insBloggerByIdResponse.getUser());
        }catch (Exception e){
           log.error("spider ins blogger by id error. msg: {}",e.getMessage());
           return;
        }

        //根据username爬取信息
        if(Boolean.FALSE.equals(insBlogger.getIsPrivate()) && Boolean.TRUE.equals(insBlogger.getIsBusiness()) && !StringUtils.isEmpty(insBlogger.getUsername()) && StringUtils.isEmpty(insBlogger.getCategoryName())
            && (insBlogger.getFollowerCount()>=collectorBloggerConfig.getInsBloggerMainCollectFollowerCount() || !collectorBloggerConfig.isInsBloggerCollectUsernameDisabled())
        ){
            try {
                InsBloggerByUsernameResponse insBloggerByUsernameResponse = insBloggerSpiderClient.getInsBloggerByUserName(insBlogger.getUsername());

                if(insBloggerByUsernameResponse != null
                        && insBloggerByUsernameResponse.getGraphql() != null
                        && insBloggerByUsernameResponse.getGraphql().getUser() != null){
                    insBlogger.setCategoryName(insBloggerByUsernameResponse.getGraphql().getUser().getCategoryName());
                    // 如果CategoryName没有就用CategoryEnum
                    if(StringUtils.isEmpty(insBlogger.getCategoryName())
                            && !StringUtils.isEmpty(insBloggerByUsernameResponse.getGraphql().getUser().getCategoryEnum())){
                        String category = insDicCategoryEnumDao.findCategoryByEnum(insBloggerByUsernameResponse.getGraphql().getUser().getCategoryEnum());
                        if(StringUtils.isEmpty(category)){
                            insBlogger.setCategoryName(category);
                        }
                    }
                    insBlogger.setBusinessCategoryName(insBloggerByUsernameResponse.getGraphql().getUser().getBusinessCategoryName());
                    insBlogger.setFbid(Long.parseLong(insBloggerByUsernameResponse.getGraphql().getUser().getFbid()));
                }

            }catch (Exception e){
                log.error("spider ins blogger by username error, msg :{}",e.getMessage());
            }
        }

        if(!StringUtils.isEmpty(insBlogger.getCategoryName())){
            String overallCategory = insDicCategoryDao.findOverallCategoryByCategory(insBlogger.getCategoryName());
            insBlogger.setOverallCategoryName(overallCategory);
        }

        // 判断是否是大量关注账户
        if(insBlogger.getFollowingCount() > collectorBloggerConfig.getZombieFollowingCount()){
            insBlogger.setIsMassiveFollowing(true);
        }else{
            // 判断是否是可疑账户
            insBlogger.setIsZombie(checkIsZombie(insBlogger));
        }

        // 更新标记字段
        insBlogger.setLastCollectTime(System.currentTimeMillis() / 1000);

        insBloggerDao.updateByUid(insBlogger);

        //如果粉丝数大于一定量，发送插入大博主的消息
        if(insBlogger.getFollowerCount() >= collectorBloggerConfig.getInsBloggerMainCollectFollowerCount()){
            sendInsertBloggerMainMessage(insBloggerMessages);
        }

        if ((System.currentTimeMillis() - start) / 1000 > collectorBloggerConfig.getDebugTimeLimit()) {
            log.info("InsBlogger update uid:{} use time:{}s", insBloggerMessages.getUid(),(System.currentTimeMillis() - start) / 1000);
        }
    }

    /**
     * 判断博主是否是僵尸粉
     * @param insBlogger
     * @return
     */
    public boolean checkIsZombie(InsBlogger insBlogger){
        if(insBlogger.getIsAnonymousPic()){
            return true;
        }
        if(insBlogger.getFollowerCount() == 0){
            return true;
        }
        if(insBlogger.getMediaCount() == 0){
            return true;
        }
        if(insBlogger.getFollowingCount() == 0){
            return true;
        }
        return false;
    }



    /**
     * 发送插入大博主队列消息
     * @param insBloggerMessage
     */
    public void sendInsertBloggerMainMessage(InsBloggerMessage insBloggerMessage){
        insBloggerMessage.setEventTime(System.currentTimeMillis());
        insBloggerMessage.setRetryLimitTimes(collectorBloggerConfig.getInsBloggerMainRetryTimes());
        insBloggerMessage.setRetryTimes(collectorBloggerConfig.getInsBloggerMainRetryTimes());
        kafkaTemplateInsBlogger.send(collectorBloggerConfig.getInsBloggerMainCollectOnlineTopic(),insBloggerMessage);
    }
}
